import java.net.*;
import java.io.*;
import java.util.HashMap;
import java.util.concurrent.CopyOnWriteArrayList;

//implements reliable broadcast
public class ReliableChannel extends BroadcastChannel{
    private ChatClientRecord owner;//the owner of this channel (the one who sends on it)
    private ApplicationListener listener;//the application who wants to receive the messages
    private CopyOnWriteArrayList<ChatClientRecord> group;//list of members of this channel
    private HashMap<String, CopyOnWriteArrayList<ChatMessage>> totalHistory;
    private MessageReceiver receiver;//listens for incoming messages
    private long messageCounter;//creates the unique message ID for messages originating on this channel

    //Initialize a reliable broadcast channel with the initial member with record 'rec'
    public ReliableChannel(ChatClientRecord rec, ApplicationListener l){
        owner = rec;//initialize the channel owner (sender)
        listener = l;//initialze the application listener
        messageCounter = 0;//initialize the unique message ID counter to 0

        //initialize the chat group
        group = new CopyOnWriteArrayList<ChatClientRecord>();
        group.add(rec);

        //initialize the received message history
        totalHistory = new HashMap<String, CopyOnWriteArrayList<ChatMessage>>();

        //start listening for incoming messages
        receiver = new MessageReceiver(rec.port, this);
        receiver.start();
    }

    //add a member to the channel
    //Returns true if successful
    public boolean addMember(ChatClientRecord member){
        if(!group.contains(member)){
            return group.add(member);
        }
        return false;
    }

    //remove a member from the channel
    //Returns true if successful
    public boolean removeMember(ChatClientRecord member){
        return group.remove(member);
    }

    //Return the current list of members
    public CopyOnWriteArrayList<ChatClientRecord> getMembers(){
        return group;
    }

    //broadcast a message
    //Return true if successful
    public boolean sendMessage(ChatMessage m){
        //set the unique message identifier only if it hasn't been set
        //(this means we don't give forwarded messages new identifiers
        if(m.messID == -1){
            System.out.println("Warning: changing messID to "+messageCounter);
            m.messID = messageCounter;
            messageCounter++;
        }
        //send the message to everyone in the group including yourself
        for(int i = 0; i < group.size(); i++){
            ChatClientRecord rec = group.get(i);
            new MessageSender(rec.ip, rec.port, m).start();
        }

        //the message was sent to yourself, so it'll be 'received' by the client then

        return true;
    }

    public void receive(ChatMessage m){
        //check to see if we've seen the message before or not
        synchronized (totalHistory){
            //get our local copy of the message sender's history
            CopyOnWriteArrayList<ChatMessage> clientHistory = totalHistory.get(m.clientRec.name);

            //if we don't have a history for them yet
            if(clientHistory == null){
                //add a history
                clientHistory = new CopyOnWriteArrayList<ChatMessage>();
                //add the message to the history
                clientHistory.add(m);
                totalHistory.put(m.clientRec.name, clientHistory);

                //send it to everyone you know
                sendMessage(m);

                //deliver message to self
                listener.receive(m);
            }else{
                //if we haven't already seen this message
                if(!clientHistory.contains(m)){
                    //add the message to the history
                    clientHistory.add(m);

                    //send it to everyone you know
                    sendMessage(m);

                    //deliver message to self
                    listener.receive(m);
                }//else ignore the message
            }
        }
    }

    private class MessageSender extends Thread {
        private String ip;
        private long port;
        ChatMessage m;

        public MessageSender(String ip, long port, ChatMessage m){
            this.ip = ip;
            this.port = port;
            this.m = m;
        }

        public void run(){
            try{
                //open connection
                Socket sock = new Socket(ip, (int)port);    
                ObjectOutputStream out = new ObjectOutputStream(sock.getOutputStream());
                //write message
                out.writeObject(m);
                //close connection
                sock.close();
                out.close();
            } catch (UnknownHostException e) {
                System.err.println("Could not connect to " + m.clientRec.name + " at " + ip + ":" + port);
            } catch (IOException e) {
                System.err.println("Error: " + e.getMessage() + " to " + ip + ":" + port);
            }
        }
    }

    private class MessageReceiver extends Thread {
        private ServerSocket sock;
        private ApplicationListener app;

        public MessageReceiver(long port, ApplicationListener app){
            //register application listener for callbacks
            this.app = app;
            try{
                sock = new ServerSocket((int)port);
            }catch(IOException e){
                System.err.println("MessageReceiver Error: "+e.getMessage());
                System.exit(1);
            }
        }

        public void run(){
            while(true){
                try{
                    //open connection
                    Socket client = sock.accept();
                    ObjectInputStream in = new ObjectInputStream(client.getInputStream());
                    //System.out.println("Accepted connection from " +client.getInetAddress().getHostName() + ":" + client.getPort());

                    //read message
                    ChatMessage message = (ChatMessage)in.readObject();
                    app.receive(message);

                    //close connection
                    in.close();
                    client.close();
                }catch(IOException e){
                    System.err.println("Error: "+e.getMessage());
                }catch(ClassNotFoundException e){
                    System.err.println("Error: "+e.getMessage());
                    System.exit(1);//if we can't find the class then we can't continue in the client
                }
            }
        }
    }
}
